#include "udp_server.h"

namespace safe_udp
{
    UdpServer::UdpServer()
    {
        sliding_window_ = std::make_unique<SlidingWindow>();

        packet_statistics_ = std::make_unique<PacketStatistics>();

        sockfd_ = 0;
        smoothed_rtt_ = 20000;
        smoothed_timeout_ = 30000; // 微秒
        dev_rtt_ = 0;

        initial_seq_number_ = 0;
        start_byte_ = 0;
        ssthresh_ = 128;
        cwnd_ = 1;

        is_slow_start_ = true;
        is_cong_avd_ = false;
        is_fast_recovery_ = false;
    }

    // 服务器启动
    int UdpServer::StartServer(int port)
    {
        int sfd;
        struct sockaddr_in server_addr; // 服务端地址
        lg(Info, "Starting the webserver... port:%d", port);
        sfd = socket(AF_INET, SOCK_DGRAM, 0); // 创建套接字

        if (sfd < 0)
        {
            lg(Info, "Falied to socket!");
            exit(0);
        }
        lg(Info, "socket success");
        memset(&server_addr, 0, sizeof(server_addr));
        server_addr.sin_family = AF_INET;
        server_addr.sin_addr.s_addr = inet_addr("127.0.0.1"); // 字符串转IP
        server_addr.sin_port = htons(port);                   // 主机转网络字节序
        lg(Info, "start bind");
        if (bind(sfd, (sockaddr *)&server_addr, sizeof(server_addr)) < 0)
        {
            lg(Error, "binding error!!");
            exit(1);
        }
        lg(Info, "bind success");
        // 服务器启动完成
        // lg(Info, "**server Bind set to addr: %s", server_addr.sin_addr);
        // lg(Info, "**server Bind set to port: %d", server_addr.sin_port);
        lg(Info, " Started successfully");
        sockfd_ = sfd;
        return sfd;
    }

    // 监听客户端请求
    char *UdpServer::GetRequest(int client_sockfd)
    {
        char *buffer = reinterpret_cast<char *>(calloc(MAX_PACKET_SIZE, sizeof(char))); // 分配1472字节

        // 记录客户端的结构体
        struct sockaddr_in client_address;
        socklen_t addr_size;
        memset(buffer, 0, MAX_PACKET_SIZE);
        addr_size = sizeof(client_address);

        // 阻塞式监听
        while (recvfrom(client_sockfd, buffer, MAX_PACKET_SIZE, 0,
                        (struct sockaddr *)&client_address, &addr_size) <= 0)
            ;

        // 有客户端连接
        lg(Info, "**Request received is: %s", buffer);
        // 记录客户端地址
        cli_address_ = client_address;

        return buffer;
    }

    // 打开文件
    bool UdpServer::OpenFile(const std::string &file_name)
    {
        lg(Info, "打开文件流，文件名字:%s", file_name);
        // 打开文件
        file_.open(file_name.c_str(), std::ios::in);

        // 打开是否成功
        if (!this->file_.is_open()) // 打开失败
        {
            lg(Info, "文件：%s 打开失败", file_name);
            return false;
        }
        else // 打开成功
        {
            lg(Info, "文件：%s 打开成功", file_name);
            return true;
        }
    }
    // 开始文件传输
    void UdpServer::StartFileTransfer()
    {
        lg(Info, "开始文件传输");

        file_.seekg(0, std::ios::end); // 定位到文件末尾
        file_length_ = file_.tellg();  // 获取文件长度
        file_.seekg(0, std::ios::beg); // 定位到文件开头
        // 开始传输
        send();
    }
    // 发送错误信息
    void UdpServer::SendError()
    {
        std::string error("FILE NOT FOUND");
        // 发送错误信息给客户端
        sendto(sockfd_, error.c_str(), error.size(), 0,
               (struct sockaddr *)&cli_address_, sizeof(cli_address_));
    }

    // 发送过程
    void UdpServer::send()
    {
        lg(Info, "Entering send()");
        int sent_count = 1;       // 当前发送了多少数据包
        int sent_count_limit = 1; // 当前窗口允许发送的最大数据包数量

        struct timeval process_start_time; // 起始时间
        struct timeval process_end_time;   //
        gettimeofday(&process_start_time, nullptr);

        // 如果滑动窗口没有任何数据
        if (sliding_window_->last_packet_sent_ == -1)
        {
            start_byte_ = 0;
        }

        while (start_byte_ <= file_length_)
        {
            fd_set rfds;
            struct timeval tv;
            int res; // 记录就绪的文件数量

            sent_count = 1;
            sent_count_limit = std::min(rwnd_, cwnd_);

            lg(Info, "send start!!!");
            lg(Info, " 发送前接收窗口大小:%d", rwnd_);
            lg(Info, " 发送前拥塞窗口大小:%d", cwnd_);
            lg(Info, "窗口使用量: %d", sliding_window_->last_packet_sent_ - sliding_window_->last_acked_packet);

            // 根据窗口数量允许发送
            while (sliding_window_->last_packet_sent_ - sliding_window_->last_acked_packet <= std::min(rwnd_, cwnd_) && sent_count <= sent_count_limit)
            {
                // 发送数据
                send_packet(start_byte_ + initial_seq_number_, start_byte_);

                // 发送完判断是处于哪种状态下发送的，就把该种状态下数据包发送数量+1
                if (is_slow_start_)
                {
                    packet_statistics_->slow_start_packet_sent_count_++;
                }
                else if (is_cong_avd_)
                {
                    packet_statistics_->cong_avd_packet_sent_count_++;
                }

                // 每次发送完1个数据包，把起始字节向后移动1460
                start_byte_ = start_byte_ + MAX_DATA_SIZE;

                // 如果移动1460字节后，起始字节代表的位置超过了总文件长度
                // （说明这个文件特别小，小于1460字节）
                // 也说明发送完的数据包就是最后1个数据包，移动1460字节后，已经没有文件数据需要发送了
                if (start_byte_ > file_length_)
                {
                    lg(Info, " 没有遗留的文件数据需要发送,不能再发送，跳出发送循环");
                    break;
                }

                sent_count++; // 发送一个数据包，当前发送数据包数量++
            }

            lg(Info, "已发送完所有文件数据");
            FD_ZERO(&rfds);         // 事前清空
            FD_SET(sockfd_, &rfds); // 把监听的套接字设置进位图里

            // 给tv赋值，smoothed_timeout是以微妙为单位
            tv.tv_sec = (int64_t)smoothed_timeout_ / 1000000; // 赋值超时时间的秒部分
            // tv.tv_sec实际上表示的是秒数对应的微秒数，
            // 所以实际上是在减去已经计算为秒的那部分微秒数，从而得到剩余的微秒数。
            tv.tv_usec = smoothed_timeout_ - tv.tv_sec * 1000000; // 赋值超时时间微妙的部分

            lg(Info, "Select 开始，超时重传时间为：%d", smoothed_timeout_);
            lg(Info, "进入死循环,循环做返回应答，除非超时或者发送完数据包，才可以跳出循环");
            while (true)
            {
                res = select(sockfd_ + 1, &rfds, nullptr, nullptr, &tv);
                if (res == -1)
                {
                    lg(Error, "select 失败");
                }
                else if (res > 0)
                {
                    lg(Info, "select 成功");
                    wait_for_ack(); // 等待应答

                    // 每次发送完，拥塞窗口都有所变化
                    if (cwnd_ < ssthresh_) // 进入拥塞避免状态
                    {
                        lg(Info, "进入拥塞避免状态");
                        is_cong_avd_ = true;
                        is_slow_start_ = false;
                    }

                    if (sliding_window_->last_acked_packet == sliding_window_->last_packet_sent_) // 发送完所有数据包
                    {
                        lg(Info, "发送完当前滑动窗口所有数据包，更新拥塞窗口值");
                        if (is_slow_start_)
                        {
                            cwnd_ = cwnd_ * 2; // 慢启动状态，拥塞窗口指数增长1次
                            lg(Info, "如果是慢启动状态发送的1次数据包，拥塞窗口指数级增大1次");
                        }
                        else if (is_cong_avd_)
                        {
                            cwnd_ = cwnd_ + 1; // 拥塞避免状态，拥塞窗口值+1
                            lg(Info, "如果是拥塞避免状态下发送的1次数据包，拥塞窗口值+1,增长1次");
                        }
                        lg(Info, "可以跳出死循环");
                        break;
                    }
                }
                else // res==0,说明没有套接字读事件就绪
                {
                    lg(Info, "res=0 没有套接字读事件就绪，意味着超时,进入慢启动状态");
                    ssthresh_ = cwnd_ / 2; // 慢启动状态，把慢启动阈值重置为原来最大拥塞窗口的一半
                    if (ssthresh_ < 1)
                    {
                        ssthresh_ = 1;
                    }
                    cwnd_ = 1; // 慢启动状态，拥塞窗口重置为1

                    // 更改其他状态标志位
                    if (is_fast_recovery_)
                    {
                        is_fast_recovery_ = false;
                    }
                    is_slow_start_ = true; // 设置慢启动状态
                    is_cong_avd_ = false;

                    // 准备快速重传,先找到所有未收到ack的数据包
                    for (int i = sliding_window_->last_acked_packet + 1;
                         i <= sliding_window_->last_packet_sent_; i++)
                    {
                        int retransmit_start_byte = 0; // 需要重传的数据包的起始字节
                        if (sliding_window_->last_acked_packet != -1)
                        {
                            // 需要重传数据包的起始字节从已发送已确认数据包的下一个数据包的起始字节开始
                            retransmit_start_byte = sliding_window_->sliding_window_buffers_[sliding_window_->last_acked_packet]
                                                        .first_byte_ +
                                                    MAX_DATA_SIZE;
                        }
                        lg(Info, "需要重传的数据包的序号为%d",
                           retransmit_start_byte + initial_seq_number_);

                        // 重发数据包
                        retransmit_segment(retransmit_start_byte);
                        packet_statistics_->retransmit_count_++; // 重传数据包数量+1
                        lg(Info, "重传起始字节为%d", retransmit_start_byte);
                    }
                    break;
                }
            }
            lg(Info, "select结束");
            lg(Info, "目前发送的字节数，起始字节为%d,文件长度为%d", start_byte_, file_length_);
        }
        // 记录发送结束时间
        gettimeofday(&process_end_time, nullptr);

        // 总共花费时间
        int64_t total_time_ =
            (process_end_time.tv_sec * 1000000 + process_end_time.tv_usec) -
            (process_start_time.tv_sec * 1000000 + process_start_time.tv_usec);

        // 包裹发送数量=慢启动发送数量+拥塞避免发送数量
        int total_packet_sent = packet_statistics_->slow_start_packet_sent_count_ + packet_statistics_->cong_avd_packet_sent_count_;

        lg(Info, "=============================================");
        lg(Info, "总共花费时间:%f秒", total_time_ / pow(10, 6)); // pow表示10的6次方
        lg(Info, "慢启动阶段发送数量:%d", packet_statistics_->slow_start_packet_sent_count_);
        lg(Info, "慢启动阶段丢包概率为%f\%", ((float)packet_statistics_->slow_start_packet_sent_count_ / total_packet_sent) * 100);
        lg(Info, "拥塞避免阶段发送数量:%d", packet_statistics_->cong_avd_packet_sent_count_);
        lg(Info, "拥塞避免阶段丢包概率为:%f\%", ((float)packet_statistics_->cong_avd_packet_sent_count_ / total_packet_sent) * 100);
        lg(Info, "重发次数：%d", packet_statistics_->retransmit_count_);
    }

    // 定位发送哪个数据包
    void UdpServer::send_packet(int seq_number, int start_byte)
    {
        bool lastPacket = false;
        int dataLength = 0;

        // 剩最后1个包裹
        if (file_length_ < start_byte + MAX_DATA_SIZE)
        {
            lg(Info, "最后1个包裹即将发送");
            dataLength = file_length_ - start_byte; // 剩余数据长度
            lastPacket = true;
        }
        else
        {
            dataLength = MAX_DATA_SIZE; // 数据长度为正常的1460字节
        }

        struct timeval time;
        gettimeofday(&time, nullptr); // 获取当前时间

        // 当前数据包的起始字节小于末数据包的起始字节,说明当前数据包在滑动窗口范围内
        if (sliding_window_->last_packet_sent_ != -1 &&
            start_byte <
                sliding_window_->sliding_window_buffers_[sliding_window_->last_packet_sent_]
                    .first_byte_)
        {
            // 遍历滑动窗口所有数据包,截至条件为小于末数据包
            for (int i = sliding_window_->last_acked_packet + 1;
                 i < sliding_window_->last_packet_sent_; i++)
            {
                // 如果当前数据包的起始字节等于起始字节，当前数据包就是要发送的数据包
                if (sliding_window_->sliding_window_buffers_[i].first_byte_ == start_byte_)
                {
                    sliding_window_->sliding_window_buffers_[i].time_sent_ = time; // 记录当前发送时间
                    break;
                }
            }
        }
        else // 当前数据包起始字节大于等于滑动最后1个数据包的起始字节
        // 说明当前数据包属于未发送未确认区的第一个数据包或者滑动窗口的最后1个数据包
        {
            SlidWinBuffer slidingWindowBuffer; // 定义1个新的数据缓冲区
            slidingWindowBuffer.first_byte_ = start_byte;
            slidingWindowBuffer.data_length = dataLength;

            slidingWindowBuffer.seq_num_ = initial_seq_number_ + start_byte; // 序号=初始序号+字节数（起始字节表征的就是字节数）

            struct timeval time;
            gettimeofday(&time, nullptr);

            slidingWindowBuffer.time_sent_ = time; // 记录发送时间
            // （认为）发送完，更新滑动窗口末数据包指针
            sliding_window_->last_packet_sent_ =
                sliding_window_->AddTobufffer(slidingWindowBuffer); // 向滑动窗口增添新的数据包
        }

        read_file_and_send(lastPacket, start_byte, start_byte + dataLength);
    }

    // 重发文件的哪部分数据
    void UdpServer::read_file_and_send(bool fin_flag, int start_byte, int end_byte)
    {
        int datalength = end_byte - start_byte; // 需要重传的数据长度
        // 如果文件剩余数据长度<计算出的需要重传的数据长度
        if (file_length_ - start_byte < datalength)
        {
            datalength = file_length_ - start_byte;
            fin_flag = true;
        }

        char *fileData = reinterpret_cast<char *>(calloc(datalength, sizeof(char)));
        if (!file_.is_open()) // 如果文件没有被打开
        {
            lg(Info, "文件打开失败");
            return;
        }
        // 将文件指针移动到指定位置
        file_.seekg(start_byte);
        // 从文件中读取datalength长度数据到fileData中
        file_.read(fileData, datalength);

        // 准备拼接成safe_udp报文格式
        DataSegment *data_segment = new DataSegment();
        data_segment->seq_number_ = start_byte + initial_seq_number_; // 设置序列号
        data_segment->ack_number_ = 0;                                // 设置确认号
        data_segment->ack_flag_ = false;                              // 设置ACK标志位
        data_segment->fin_flag_ = fin_flag;                           // 设置FIN标志位
        data_segment->length_ = datalength;                           // 设置数据长度
        data_segment->data_ = fileData;                               // 设置报文有效载荷

        // 正式发送数据包
        send_data_segment(data_segment);
        lg(Info, "当前发送数据包的序号为：%d", data_segment->seq_number_);

        free(fileData);
        free(data_segment);
    }

    // 重发数据段
    void UdpServer::send_data_segment(DataSegment *data_segment)
    {
        // 数据序列化（将数据拼接成报文格式），返回值为序列化好的一串字符
        char *datagramChars = data_segment->SerializeToCharArray();

        // 发送数据到客户端，发送一个数据包大小，不管数据长度小于1472字节还是等于1472字节，都按1472字节发送
        sendto(sockfd_, datagramChars, MAX_PACKET_SIZE, 0,
               (struct sockaddr *)&cli_address_, sizeof(cli_address_));
        // 释放字符数组占用的内存
        free(datagramChars);
    }

    // 等待确认，根据确认情况（丢包或全部确认收到）执行对应的逻辑
    void UdpServer::wait_for_ack()
    {
        unsigned char buffer[MAX_PACKET_SIZE]; // 缓冲区
        memset(buffer, 0, MAX_PACKET_SIZE);

        // 记录客户端信息的变量
        socklen_t addr_size;
        struct sockaddr_in client_address;
        addr_size = sizeof(client_address);
        int n = 0;
        int ack_number;
        // 从网络套接字接收的信息存在了buffer缓冲区里，阻塞式等待
        while ((n = recvfrom(sockfd_, buffer, MAX_PACKET_SIZE, 0,
                             (struct sockaddr *)&client_address, &addr_size)) <= 0)
        {
        };

        DataSegment ack_segment;                         // 序列化和反序列化用的
        ack_segment.DeserializeToDataSegment(buffer, n); // 将报文反序列化

        // 新建一个缓冲区对象，赋值为滑动窗口已发送已确认数据缓冲区（起始指针）
        SlidWinBuffer last_packet_acked_buffer = sliding_window_->sliding_window_buffers_[sliding_window_->last_acked_packet];

        // 如果反序列化后提取到的ack标志位为真，表示这是1个确认应答报文，还可能是捎带应答，
        // 如果不是应答报文，那就是服务端给客户端发的文件数据的数据包
        if (ack_segment.ack_flag_)
        {
            // 反序列化后的确认序号和滑动窗口数据缓冲区的已发送已确认的数据包中序号相等
            // 报文的确认序号和滑动窗口数据缓冲区的已发送已确认的数据包序号做对比
            // 这是检查是否收到了对当前发送窗口中最旧数据包的确认。
            // 是，则表示我上一个报文你已经收到，因为你给我ack应答了，并对该报文做1次重传计数
            if (ack_segment.ack_number_ == sliding_window_->send_base_)
            {
                lg(Info, "收到1个应答报文，确认序号为:%d 做1次重复收到报文计数", ack_segment.ack_number_);

                sliding_window_->dup_ack_++; // 收到一次报文，计数+1

                // 如果收到同样的ack为3，做快速重传
                if (sliding_window_->dup_ack_ == 3)
                {
                    packet_statistics_->retransmit_count_++; // 重传次数+1
                    lg(Info, "收到3次同样ack，需要重传的数据包的序号为：%d", ack_segment.ack_number_);

                    // 快速重传的接口
                    retransmit_segment(ack_segment.ack_number_ - initial_seq_number_);

                    sliding_window_->dup_ack_ = 0; // 重传计数置0
                    if (cwnd_ > 1)
                    {
                        cwnd_ = cwnd_ / 2; // 拥塞窗口变为原来的1半
                    }
                    ssthresh_ = cwnd_;        // 慢启动阈值也变为更新后的拥塞窗口大小
                    is_fast_recovery_ = true; // 处于快速恢复状态
                    // LOG(INFO) << "Change to fast Recovery ack_segment->ack_number:"
                    //           << ack_segment->ack_number_;
                }
            }

            // 如果报文中的确认序号表示大于你发送的
            // 意味着之前发送的多个数据包已被成功接收，并且发送窗口可以向前移动。
            else if (ack_segment.ack_number_ > sliding_window_->send_base_)
            {
                if (is_fast_recovery_) // 如果处于快速恢复状态
                {
                    // LOG(INFO) << "Change to Cong Avoidance from fast recovery recv ack:"
                    //           << ack_segment->ack_number_;
                    cwnd_++;
                    is_fast_recovery_ = false; // 退出快速恢复状态
                    is_cong_avd_ = true;       // 处于拥塞避免状态
                    is_slow_start_ = false;    // 不处于慢启动状态
                }

                // 无论是否处于快速恢复状态，都重置重复ACK的计数（sliding_window_->dup_ack_ = 0），
                // 因为此时已经收到了一个有效的ACK，表明之前的包已经被对方成功接收
                sliding_window_->dup_ack_ = 0;

                sliding_window_->send_base_ = ack_segment.ack_number_; // 更新发送基础序号

                if (sliding_window_->last_acked_packet == -1)
                {
                    // 更新已发送已确认（滑动窗口起始指针）
                    sliding_window_->last_acked_packet = 0;

                    // 最后一个已发送已确认缓冲区更新未滑动窗口内的最后一个已发送已确认数据包
                    last_packet_acked_buffer =
                        sliding_window_
                            ->sliding_window_buffers_[sliding_window_->last_acked_packet];
                }
                // 当前确认序号=已确认的最后一个数据包序号+该数据包的长度
                ack_number = last_packet_acked_buffer.seq_num_ +
                             last_packet_acked_buffer.data_length;

                // 遍历滑动窗口中的所有数据包，直到找到包含或超过ACK确认序列号的数据包
                while (ack_number < ack_segment.ack_number_)
                {
                    sliding_window_->last_acked_packet++; // 已发送已确认（起始指针）+1，指向下一个数据包

                    // 获取滑动窗口内已发送已确认（起始指针）数据包
                    last_packet_acked_buffer =
                        sliding_window_
                            ->sliding_window_buffers_[sliding_window_->last_acked_packet];
                    // 更新确认序号
                    ack_number = last_packet_acked_buffer.seq_num_ +
                                 last_packet_acked_buffer.data_length;
                }

                struct timeval startTime = last_packet_acked_buffer.time_sent_;
                struct timeval endTime;
                gettimeofday(&endTime, NULL); // 获取当前时间作为结束时间

                // LOG(INFO) << "seq num of last_acked "
                //           << last_packet_acked_buffer.seq_num_;
                calculate_rtt_and_time(startTime, endTime); // 计算RTT（往返时延）
            }
            // LOG(INFO) << "sliding_window_->lastAckedPacket"
            //           << sliding_window_->last_acked_packet_;
        }
    }

    // 计算超时时间
    void UdpServer::calculate_rtt_and_time(struct timeval start_time, struct timeval end_time)
    {
        // 如果起始时间为0，则直接返回
        if (start_time.tv_sec == 0 && start_time.tv_usec == 0)
        {
            return;
        }
        // 本次测量到的往返时间
        long sample_rtt = (end_time.tv_sec * 1000000 + end_time.tv_usec) -
                          (start_time.tv_sec * 1000000 + start_time.tv_usec);
        // 使用指数平滑算法更新平滑RTT
        // smoothed_RTT = (1 - α) * smoothed_RTT + α * sample_RTT
        smoothed_rtt_ = smoothed_rtt_ + 0.125 * (sample_rtt - smoothed_rtt_);

        // 使用指数平滑算法更新RTT的偏差
        // deviation = (1 - β) * deviation + β * |sample_RTT - smoothed_RTT|
        dev_rtt_ = 0.75 * dev_rtt_ + 0.25 * (abs(smoothed_rtt_ - sample_rtt));

        // 计算平滑超时时间
        smoothed_timeout_ = smoothed_rtt_ + 4 * dev_rtt_;

        // 如果平滑超时时间大于1秒，则重新计算超时时间为随机数
        if (smoothed_timeout_ > 1000000)
        {
            smoothed_timeout_ = rand() % 30000;
        }
    }

    // 哪个数据包需要快重传
    void UdpServer::retransmit_segment(int index_number)
    {
        // 滑动窗口起始指针<滑动窗口末指针
        for (int i = sliding_window_->last_acked_packet + 1;
             i < sliding_window_->last_packet_sent_; i++)
        {
            // 判断当前数据段的起始字节号是否与需要重传的起始字节号相等，
            // 如果找到了该起始字节号，则该数据段是需要重传的
            if (sliding_window_->sliding_window_buffers_[i].first_byte_ ==
                index_number)
            {
                struct timeval time;
                gettimeofday(&time, NULL); // 获取当前时间

                // 记录当前数据段的发送时间
                sliding_window_->sliding_window_buffers_[i].time_sent_ = time;
                break; // 跳出循环
            }
        }
        // 当前数据包不是文件数据的最后一个数据包 需要重传数据包的起始字节序号，终止字节序号
        read_file_and_send(false, index_number, index_number + MAX_DATA_SIZE);
    }
}


